package somebak;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class BitmapTest2 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 1. 创建映射表，映射kafka明细层的行为日志
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                                "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_province STRING,                     "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     proc_time AS proctime(),                      "
                        + "     row_time AS to_timestamp_ltz(event_time,3),   "
                        + "     watermark for row_time as row_time - interval '0' second   "
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                         "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup2',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           "
        );

        //tenv.executeSql("select * from dwd_kafka").print();


        // 2. 创建映射表，映射doris中的聚合表
        tenv.executeSql(
                "CREATE TABLE sink_doris(                           " +
                        "    dt       DATE,                                 " +
                        "    time_60m STRING,                               " +
                        "    time_30m STRING,                               " +
                        "    time_10m STRING,                               " +
                        "    time_m   STRING,                               " +
                        "    release_channel STRING,                        " +
                        "    device_type STRING,                            " +
                        "    gps_province STRING,                           " +
                        "    gps_city STRING,                               " +
                        "    gps_region STRING,                             " +
                        "    page_type STRING,                              " +
                        "    page_service STRING,                           " +
                        "    page_url STRING,                               " +
                        "    is_new int,                               " +
                        "    pv_amt BIGINT,                                 " +
                        "    session_cnt  BIGINT,                           " +
                        "    user_id  int                                  " +
                        ") WITH (                                           " +
                        "   'connector' = 'doris',                          " +
                        "   'fenodes' = 'doitedu:8030',                     " +
                        "   'table.identifier' = 'dws.tfc_overview_u_m',    " +
                        "   'username' = 'root',                            " +
                        "   'password' = 'root',                            " +
                        "   'sink.label-prefix' = 'doris_label4',           " +
                        "   'sink.properties.columns' = 'dt,time_60m,time_30m,time_10m,time_m,release_channel,device_type,gps_province,gps_city,gps_region,page_type,page_service,page_url,is_new,pv_amt,session_cnt,user_id,user_id=to_bitmap(user_id)'             " +
                        ")                                                  "
        );

        // 3. 计算逻辑sql
        tenv.createTemporaryFunction("time_trunc", TimeStampTruncate2.class);
        tenv.executeSql(
                " create temporary view res as                                                                                          " +
                        " with tmp as (                                                                                                  " +
                        " select                                                                                                         " +
                        "   to_date(date_format(to_timestamp_ltz(event_time,3),'yyyy-MM-dd')) as dt                                      " +
                        "   ,time_trunc(event_time,60) as time_60m                                                                       " +
                        "   ,time_trunc(event_time,30) as time_30m                                                                       " +
                        "   ,time_trunc(event_time,10) as time_10m                                                                       " +
                        "   ,time_trunc(event_time,1)  as time_01m                                                                       " +
                        "   ,cast( user_id  as int) as user_id                                                                            " +
                        "   ,if(date_format(register_time,'yyyy-MM-dd')< CURRENT_DATE,0,1) as is_new              " +
                        "   ,session_id                                                                                                  " +
                        "   ,release_channel                                                                                             " +
                        "   ,device_type                                                                                                 " +
                        "   ,gps_province                                                                                                " +
                        "   ,gps_city                                                                                                    " +
                        "   ,gps_region                                                                                                  " +
                        "   ,page_type                                                                                                   " +
                        "   ,page_service                                                                                                " +
                        "   ,regexp_extract(properties['url'],'^(.*\\.html).*?') as page_url                                             " +
                        "   ,row_time                                                                                                    " +
                        " from dwd_kafka                                                                                                 " +
                        " where event_id='page_load'                                                                                     " +
                        " )                                                                                                              " +
                        "                                                                                                                " +
                        " SELECT                                                                                                         " +
                        "   dt" +
                        "   ,time_60m" +
                        "   ,time_30m" +
                        "   ,time_10m" +
                        "   ,time_01m                                                                      " +
                        "   ,release_channel                                                                                             " +
                        "   ,device_type                                                                                                 " +
                        "   ,gps_province                                                                                                " +
                        "   ,gps_city                                                                                                    " +
                        "   ,gps_region                                                                                                  " +
                        "   ,page_type                                                                                                   " +
                        "   ,page_service                                                                                                " +
                        "   ,page_url                                                                                                    " +
                        "   ,is_new                                                                                                    " +
                        "   ,count(1) as pv_amt                                                                                          " +
                        "   ,count(distinct session_id) as session_cnt                                                                                          " +
                        "   ,user_id                                                                                                            " +
                        " FROM TABLE(                                                                                                    " +
                        "    TUMBLE(TABLE tmp, DESCRIPTOR(row_time),INTERVAL '5' MINUTE)                                                 " +
                        " )                                                                                                              " +
                        " GROUP BY                                                                                                       " +
                        "   window_start,                                                                                                " +
                        "   window_end,                                                                                                  " +
                        "   dt,time_60m,time_30m,time_10m,time_01m                                                                      " +
                        "   ,release_channel                                                                                             " +
                        "   ,device_type                                                                                                 " +
                        "   ,gps_province                                                                                                " +
                        "   ,gps_city                                                                                                    " +
                        "   ,gps_region                                                                                                  " +
                        "   ,page_type                                                                                                   " +
                        "   ,page_service                                                                                                " +
                        "   ,page_url                                                                                                    " +
                        "   ,is_new                                                                                                    " +
                        "   ,user_id                                                                                                    "
        );


        // insert into sink_doris
        tenv.executeSql("insert into sink_doris select * from res");
        //tenv.executeSql("select * from res").print();

    }
}


